package com.chenfan.mcn.event;

import com.chenfan.mcn.event.subscriber.EventSubscriber;
import com.chenfan.common.extension.util.SmartExecutorService;
import com.chenfan.common.extension.spring.TransactionUtils;
import com.google.common.eventbus.AsyncEventBus;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.function.Supplier;

/**
 * 事件发布者
 *
 * @author wulg
 * @date 2021-11-24
 **/
@Slf4j
@Component
public class EventPublisher {

    private static AsyncEventBus asyncEventBus;

    private static final Supplier<RuntimeException> DEFAULT_SUPPLIER = () -> new RuntimeException("EventPublisher not configured bean!");

    public EventPublisher(@Autowired ObjectProvider<EventSubscriber> objectProvider, @Autowired(required = false) ExecutorService executorService) {
        if(asyncEventBus == null) {
            this.initAsyncEventBus(objectProvider, executorService);
        }
    }

    /**
     * 初始化事件总线
     * @param objectProvider
     * @param executorService
     */
    private synchronized void initAsyncEventBus(ObjectProvider<EventSubscriber> objectProvider, ExecutorService executorService){
        if(asyncEventBus != null) {
            return;
        }
        if(executorService == null) {
            executorService = SmartExecutorService.newFixedThreadPool();
        }
        asyncEventBus = new AsyncEventBus(executorService, new EventSubscriber.EventSubscriberExceptionHandler());
        objectProvider.stream().forEach(eventListener -> {
            asyncEventBus.register(eventListener);
        });
    }

    /**
     * 注册事件订阅者
     * @param eventSubscriber
     */
    public static void register(EventSubscriber eventSubscriber) {
        Optional.ofNullable(asyncEventBus).orElseThrow(DEFAULT_SUPPLIER).register(eventSubscriber);
    }

    /**
     * 发布事件
     * @param event
     * @param <T>
     */
    public static <T> void post(T event) {
        if(Objects.nonNull(event)) {
            TransactionUtils.afterCommit(() -> Optional.ofNullable(asyncEventBus).orElseThrow(DEFAULT_SUPPLIER).post(event));
        }
    }

    /**
     *
     * @param eventList
     * @param <T>
     */
    public static <T> void post(List<T> eventList) {
        if(CollectionUtils.isNotEmpty(eventList)) {
            try {
                eventList.forEach(EventPublisher::post);
            } finally {
                eventList.clear();
            }
        }
    }
}
